package com.example.flinkcourse.lesson2.stream;

import com.example.flinkcourse.lesson2.model.Event;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 事件处理器
 * 使用 DataStream API 处理事件数据
 */
public class EventProcessor {
    private static final Logger LOG = LoggerFactory.getLogger(EventProcessor.class);

    /**
     * 处理事件流
     * @param inputStream 输入事件流
     * @return 处理后的事件流
     */
    public static DataStream<Event> processEvents(DataStream<Event> inputStream) {
        LOG.info("Processing events with DataStream API...");
        
        return inputStream
            // 过滤无效事件
            .filter(event -> {
                boolean isValid = event != null && event.getId() != null;
                if (!isValid) {
                    LOG.warn("Invalid event received: {}", event);
                }
                return isValid;
            })
            
            // 添加处理时间
            .map(event -> {
                event.setTimestamp(System.currentTimeMillis());
                LOG.debug("Added processing time to event: {}", event.getId());
                return event;
            })
            
            // 按类型分组
            .keyBy(Event::getType)
            
            // 窗口处理
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            
            // 聚合处理
            .aggregate(new EventAggregator())
            
            // 设置并行度
            .setParallelism(4);
    }
}

/**
 * 事件聚合器
 * 用于在窗口内聚合事件数据
 */
class EventAggregator implements AggregateFunction<Event, Event, Event> {
    @Override
    public Event createAccumulator() {
        return new Event();
    }

    @Override
    public Event add(Event event, Event accumulator) {
        // 更新聚合结果
        if (accumulator.getId() == null) {
            accumulator.setId(event.getId());
            accumulator.setType(event.getType());
            accumulator.setTimestamp(event.getTimestamp());
            accumulator.setProperties(event.getProperties());
        }
        return accumulator;
    }

    @Override
    public Event getResult(Event accumulator) {
        return accumulator;
    }

    @Override
    public Event merge(Event a, Event b) {
        // 合并两个聚合结果
        return a.getTimestamp() > b.getTimestamp() ? a : b;
    }
} 